package qa.d20241106;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.sql.Timestamp;
import java.util.UUID;

/**
 * 2024年11月7日
 * 双流join 错误的使用了 Temporal Join 导致错误
 * 应该使用窗口join
 *
 *
 * 报错版本：
 * Event-Time Temporal Table Join requires both primary key and row time attribute in versioned table,
 * but no row time attribute can be found.
 */

/**
 * https://blog.csdn.net/Rango_lhl/article/details/125974764
 */
public class JoinTest {


  public static void main(String[] args) throws InterruptedException {


    Configuration flinkConf = new Configuration();
    flinkConf.setString("rest.port","9093");
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(flinkConf);
    env.setParallelism(1);
    StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

    env.enableCheckpointing(10 * 1000);
    env.setStateBackend(new EmbeddedRocksDBStateBackend());
    CheckpointConfig checkpointConfig = env.getCheckpointConfig();
    checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
    checkpointConfig.setCheckpointStorage("file:///flink_ckp");


    String ods1 = "CREATE TABLE ods_tb1 ( " +
        " stime STRING," +
        " name STRING," +
        " val BIGINT," +
        " rowtime AS to_timestamp(stime)," +
        " proctime as proctime()," +
        " WATERMARK FOR rowtime AS rowtime - interval '10' second" +
        ") WITH ( " +
        "  'connector' = 'kafka'," +
        "  'topic' = 'test1'," +
        "  'properties.bootstrap.servers' = 'kafka:9092'," +
        "  'properties.group.id' = '" + UUID.randomUUID().toString() +"'," +
        "  'scan.startup.mode' = 'latest-offset'," +
        "  'format' = 'json'" +
        ")";

    String ods2 = "CREATE TABLE ods_tb2 ( " +
        " stime STRING," +
        " name STRING," +
        " val BIGINT," +
        " rowtime AS to_timestamp(stime)," +
        " proctime as proctime()," +
        " WATERMARK FOR rowtime AS rowtime - interval '10' second" +
        ") WITH ( " +
        "  'connector' = 'kafka'," +
        "  'topic' = 'test2'," +
        "  'properties.bootstrap.servers' = 'kafka:9092'," +
        "  'properties.group.id' = '" + UUID.randomUUID().toString() +"'," +
        "  'scan.startup.mode' = 'latest-offset'," +
        "  'format' = 'json'" +
        ")";


    String wind1 = "CREATE VIEW ww1 AS " +
        " SELECT " +
        " name,count(1) as cnt,window_time as rowtime" +
        " ,window_start,window_end,window_time " +
        " from " +
        " table(tumble(table ods_tb1, descriptor(rowtime), interval '1' minutes))" +
        " GROUP BY name,window_start,window_end,window_time";

    String wind2 = "CREATE VIEW ww2 AS " +
        " SELECT " +
        " name,count(1) as cnt,window_time as rowtime" +
        " ,window_start,window_end,window_time " +
        " from " +
        " table(tumble(table ods_tb2, descriptor(rowtime), interval '1' minutes))" +
        " GROUP BY name,window_start,window_end,window_time";


    String joinRes = "CREATE VIEW res AS " +
        " SELECT t1.name,t1.rowtime,t1.cnt as cnt1, t2.cnt as cnt2 FROM ww1 t1 left join ww2 for system_time AS OF t1" +
        ".rowtime t2 on t1" +
        ".name=t2.name";



    String wind3 = "CREATE VIEW ww3 AS " +
        " SELECT " +
        " name,sum(cnt1) as cc1,sum(cnt2) as cc2" +
        " ,window_start,window_end,window_time " +
        " from " +
        " table(tumble(table res, descriptor(rowtime), interval '1' minutes))" +
        " GROUP BY name,window_start,window_end,window_time";

    String sinkPrint = "CREATE TABLE print (" +
        "    stime timestamp, " +
        "    name STRING, " +
        "    cc1 BIGINT," +
        "    cc2 BIGINT" +
        ") WITH (" +
        "     'connector' = 'print'" +
        ")";


    String sql = "INSERT INTO print" +
        " SELECT " +
        " window_time," +
        " name,cc1,cc2" +
        " FROM ww3";

    tableEnv.executeSql(ods1);
    tableEnv.executeSql(ods2);
    tableEnv.executeSql(sinkPrint);
    tableEnv.executeSql(wind1);
    tableEnv.executeSql(wind2);
    tableEnv.executeSql(joinRes);
    tableEnv.executeSql(wind3);
    tableEnv.executeSql(sql);


    System.out.println(new Timestamp(System.currentTimeMillis()));



  }
}
